package cn.doitedu.demo2;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;

public class RuleCalculator_01 implements RuleCalculator {


    private String ruleId = "rule-001";

    ValueState<Integer> ACNT;
    ValueState<Integer> IDX;
    ValueState<Integer> SEQ_CNT;
    final String[] eventSeq = {"Q", "C", "Y"};
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext) {

        ACNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("A_CNT", Integer.class));
        IDX = runtimeContext.getState(new ValueStateDescriptor<Integer>("IDX", Integer.class));
        SEQ_CNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("SEQ_CNT", Integer.class));

        message = new JSONObject();
        message.put("rule_id", "rule-002");
    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {
        String eventId = userAction.getEvent_id();

        /* *
         * 用户发生 A行为：3次以上
         * 用户发生  [ Q C Y ] 事件序列:  2次以上
         * 当他做 C行为时，触发规则
         */

        // A行为统计
        if (eventId.equals("A")) {
            int cnt = ACNT.value() == null ? 1 : ACNT.value() + 1;
            ACNT.update(cnt);
        }

        // [ Q C Y ] 事件序列的发生次数统计
        int idx = IDX.value() == null ? 0 : IDX.value();
        if (eventId.equals(eventSeq[idx])) {
            // 如果是，则要把idx 递增
            int newIdx = idx + 1;
            // 如果idx递增后等于3了，说明这个序列到此被完整发生过了
            if (newIdx == 3) {
                // 重置idx
                newIdx = 0;
                // 增加完整发生次数
                SEQ_CNT.update(SEQ_CNT.value() == null ? 1 : SEQ_CNT.value() + 1);
            }
            IDX.update(newIdx);
        }

        // 触发事件C行为 处理
        if (eventId.equals("C")) {
            Integer aCount = ACNT.value();
            Integer seqCount = SEQ_CNT.value();

            if (aCount != null && seqCount != null && aCount >= 3 && seqCount >= 2) {

                message.put("urser_id", userAction.getUser_id());
                message.put("hit_time", userAction.getAction_time());

                out.collect(message.toJSONString());

            }
        }
    }

    @Override
    public String ruleId() {
        return this.ruleId;
    }
}
